package hang.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * @author 航
 * @date 2024/04/12 10:06
 **/
public class ProducerTransactionTest {
    public static void main(String[] args) {

        // 创建 配置对象
        Map<String, Object> config = new HashMap<String, Object>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 对生产者的数据 K，V 进行序列化操作
        // StringSerializer.class.getName() 表示获取这个类的名称
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 开启幂等性
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        // 开启重试，重试5次
        config.put(ProducerConfig.RETRIES_CONFIG, 5);
        // 开启事务ID，事务是基于幂等性操作的  具体的可以自己定事务id
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-tx-id");


        // 创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer(config);
        // 初始化事务
        producer.initTransactions();

        try {
            // 开启事务
            producer.beginTransaction();
            for(int i = 0; i<10; i++){
                ProducerRecord<String, String> record = new ProducerRecord("test","key" + i,"value" + i);
                producer.send(record);
            }
            // 提交事务
            producer.commitTransaction();

        } catch (Exception e) {
            // 中止事务
            producer.abortTransaction();
            e.printStackTrace();
            throw new RuntimeException(e);
        } finally {
            // 关闭生产者对象
            producer.close();
        }



    }
}
